package com.rpc.core.remoting.impl;

import com.rpc.core.codec.RpcCodec;
import com.rpc.core.codec.adapter.NettyDecodeAdapter;
import com.rpc.core.codec.adapter.NettyEncodeAdapter;
import com.rpc.core.constant.BeatConstant;
import com.rpc.core.constant.ClientLoop;
import com.rpc.core.param.RpcParams;
import com.rpc.core.filter.RpcClientFilter;
import com.rpc.core.holder.rag.RegInfoEnhancer;
import com.rpc.core.holder.transmit.Request;
import com.rpc.core.holder.transmit.RequestSync;
import com.rpc.core.holder.transmit.Response;
import com.rpc.core.remoting.Client;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;

import java.util.Collection;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@ChannelHandler.Sharable
public class NettyClient extends SimpleChannelInboundHandler<Response> implements Client {

    //从2开始
    private AtomicInteger idCreator = new AtomicInteger(2);

    private ConcurrentHashMap<Integer, RequestSync> map = new ConcurrentHashMap<>();
    private Channel channel;

    private RegInfoEnhancer enhancer;

    public void setEnhancer(RegInfoEnhancer enhancer) {
        this.enhancer = enhancer;
    }

    Collection<RpcClientFilter> filters;

    private RpcParams params;

    private ApplicationContext context;

    private Logger logger = LoggerFactory.getLogger(NettyClient.class);

    public NettyClient(ApplicationContext context, RegInfoEnhancer regInfoEnhancer) {
        this.context = context;
        params = context.getBean(RpcParams.class);
        this.filters = context.getBeansOfType(RpcClientFilter.class).values();
        this.enhancer = regInfoEnhancer;
    }

    @Override
    public void sendSync(RequestSync requestSync) {
        int id = idCreator.getAndAdd(1);
        Request request = requestSync.getRequest();
        request.setId(id);

        //filter
        if (doFilter(request)) {
            map.put(id, requestSync);
            channel.writeAndFlush(request);
        }
    }

    private boolean doFilter(Request request) {


        if (filters != null && filters.size() != 0) {
            for (RpcClientFilter filter : filters) {
                int num = filter.filter(request);
                if (num == 0) {
                    return false;
                }
            }
        }

        return true;
    }


    @Override
    public void send(Request request) {
        channel.writeAndFlush(request);
    }

    @Override
    public void remove(int id) {
        map.remove(id);
        reId();
    }


    @Override
    public void start(String ip, int port) throws Exception {


        Object bean = context.getBean(params.getCodecType());

        if (!(bean instanceof RpcCodec)) {
            throw new RuntimeException("The specified encoder was not found!");
        }

        RpcCodec codec = (RpcCodec) bean;
        // group


        // init
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(ClientLoop.getLoop())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel channel) throws Exception {
                        channel.pipeline()
                                .addLast(new IdleStateHandler(0, 0, BeatConstant.BEAT_INTERVAL, TimeUnit.SECONDS))    // beat N, close if fail
                                .addLast(new NettyEncodeAdapter(codec))
                                .addLast(new NettyDecodeAdapter(codec))
                                .addLast(NettyClient.this);
                    }
                })
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000);


        this.channel = bootstrap.connect(ip, port).sync().channel();

        if (this.channel == null || !this.channel.isActive()) {
            this.stop();
        }


        logger.debug("rpc netty client  connect to server success at host:{}, port:{}", ip, port);

    }

    @Override
    public void stop() {
        if (this.channel != null && this.channel.isActive()) {
            this.channel.close();        // if this.channel.isOpen()
        }
        enhancer.deleteClient();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        logger.info("{} channel",channel.id());
        logger.info("{} channel inactive",ctx.channel().id());
        super.channelInactive(ctx);
        //通知 refInfoEN remove
        enhancer.deleteClient();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Response response) throws Exception {
        int id = response.getId();
        if(id==BeatConstant.BEAT_ID){
            logger.info("service beat pong");
            return;
        }
        RequestSync requestSync = map.get(id);
        if (requestSync != null) {
            requestSync.down(response);
        }

        reId();
    }

    private void reId() {
        if (map.size() == 0 && idCreator.get() >= params.getRequestIdMax()) {
            synchronized (map) {
                if (map.size() == 0 && idCreator.get() >= params.getRequestIdMax()) {
                    idCreator = new AtomicInteger(2);
                }
            }
        }
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error("rpc netty client caught exception", cause);
        ctx.close();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            //send beat
            send(BeatConstant.BEAT_PING);
            logger.info("rpc netty client send beat.");
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}
